package com.icarbonx.dap.notification.job

import com.google.gson.Gson
import com.icarbonx.dap.notification.msgservice.rabbitmq.{ConnUtil, MQObject}
import com.icarbonx.dap.notification.msgservice.redis.RedisEndpoint
import com.icarbonx.dap.notification.util.{DateUtil, KafkaUtil, NoticeMsg, SparkUtil}
import com.icarbonx.dap.notification.vo._
import kafka.serializer.StringDecoder
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}

import scala.collection.mutable.ArrayBuffer
import scala.math.{abs, pow, sqrt}

/**
 * Created by Lenovo on 2018/8/12.
 * 17.多发夜间低糖警报:连续两个点血糖<3.0mmol/L,5天内出现3次&0:00~6:00:low_glycemia_night_freq:每日夜间:第二天早上8点，每小时提示不超过1条
 * 22.多发空腹高糖警报:6:00&前后两个点血糖>7.8mmol/L，5内出现3次:high_glycemia_fast_freq:每天1次:触发即时发送，每小时不超过1条
 * 12.低糖:连续两个点血糖<2.8mmol/L:low_glycemia:每5min:触发及时发送. 每小时提示不超过1条
 * 13.高糖:连续两个点血糖>12.0mmol/L:high_glycemia:每5min:触发及时发送. 每小时提示不超过1条
 * 30.空腹高糖:6:00&前后两个点血糖>8.6mmol/L:high_glycemia_fast:每天1次:触发及时发送
 * 23.夜间低糖:检测到连续2点<2.8mmol/L&0:00~6:00:low_glycemia_night:每天1次:第二天早上8点回顾性检测0:00~6:00数据
 * 14.升糖过速，15.降糖过速:血糖上升/下降变化超过6.0mmol/L/h&全天:slope_glycemia_high/slope_glycemia_low:每5min:触发及时发送. 每小时提示不超过1条
 * 7.连续4个小时没有血糖数据:连续血糖仪数据中断超过4小时:miss_glycemia_0:每5min:检测到数据中断达到4小时。确认sensor 失效即停止所有CGM相关提示
 * 8.连续7.5个小时没有血糖数据:连续血糖仪数据中断超过7.5小时:miss_glycemia_1:每5min:检测到数据中断达到7.5小时，确认sensor 失效即停止所有CGM相关提示
 * 9.连续20个小时没有血糖数据:连续血糖数据中断超20小时:miss_glycemia_2:每5min:检测到数据中断达到16时， 确认sensor 失效即停止所有CGM相关提示
 * 33.低血糖：LLD>0.4&全天:low_glycemia_ratio:每天1次:随每日总结发送
 * 34.高血糖：高于目标值时间占比>50, MAGE<0.4&全天:hight_glycemia_ratio:每天1次:随每日总结发送
 * 35.血糖波动：LLD>0.4&全天:mage:每天1次:随每日总结发送
 * 多发夜间低糖，多发空腹高糖
 * lld，mage，hld
 * 共计14个警报，4个指标
 */
object BloodSugarStream {
  def main(args: Array[String]) {
    Logger.getLogger("org.apache.kafka").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.ERROR)
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    println(args.mkString("\t"))

    val Array(brokers, topicList, mqHost, mqPort, mqUsername,
    mqPassword, mqVirtualHost, redisHost, redisPort, redisPassword,
    redisTimeOut, exchangeName, mqRouteKey, checkpointDir, duration, rddDuration, writeFileST, writeFileET, appName
    ) = args

    val conf = SparkUtil.getSparkConf(appName, Array(classOf[BloodSugarRecord]))

    val sc = SparkContext.getOrCreate(conf)
    //    val checkpointDir = "/data/streaming/checkpoint/bloodsugar_stream/"

    val output_indicator_path = "/data/streaming/bloodsugar_stream/"

    val updateFunc = (values: Seq[ArrayBuffer[BloodSugarRecord]],
                      state: Option[ArrayBuffer[BloodSugarRecord]]) => {
      //保存状态  过滤今天的数据
      //      val todayTimeStamp = DateUtil.getTodayDateTime("yyyyMMdd")
      val nowTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss") - (24 * 60 * 60 * 1000 + 30 * 60 * 1000)
      //      val flag1 = nowTimeStamp - todayTimeStamp <= 45 * 60 * 1000
      val currentCount = ArrayBuffer[BloodSugarRecord]()
      values.foreach { list => currentCount ++= list }
      val previousCount = state.getOrElse(ArrayBuffer[BloodSugarRecord]())
      //      Some(currentCount ++ previousCount.filter { record =>
      //        if (flag1) true
      //        else if (record.timestamp - todayTimeStamp > 0) true
      //        else false
      //      })
      Some(currentCount ++ previousCount.filter(vo => vo.timestamp > nowTimeStamp))
    }

    def creatingFunc(): StreamingContext = {
      val ssc = new StreamingContext(sc, Seconds(duration.toInt))
      ssc.checkpoint(checkpointDir)
      val kafkaParams = KafkaUtil.getKafkaParam(brokers, appName)

      val mqConn = new MQObject(mqHost, mqPort.toInt, mqUsername, mqPassword, mqVirtualHost)
      val redisConn = new RedisEndpoint(redisHost, redisPort.toInt, redisPassword, redisTimeOut.toInt)

      val topics = topicList.split(",").toSet
      val bloodSugarMessages = KafkaUtils
        .createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc,
          kafkaParams,
          topics
        )

      val newBatchAndOld: DStream[(String, ArrayBuffer[BloodSugarRecord])] =
        bloodSugarMessages
          .filter(_._2.nonEmpty)
          .flatMap { msg =>
            try {
              println("___________ msg :" + msg._2)
              val gson = new Gson()
              val voList =
                gson.fromJson(msg._2, classOf[Array[BloodSugarRecord]])
              voList.map { vo =>
                (vo.personId.toString, ArrayBuffer[BloodSugarRecord](vo))
              }.toList
            } catch {
              case e: Exception =>
                println("____________" + e.getMessage)
                List(("", ArrayBuffer[BloodSugarRecord]()))
            }
          }
          .filter(_._1.nonEmpty)
          .updateStateByKey(updateFunc)

      //      val notNullRdd = newBatchAndOld.filter(_._2.size > 0)

      newBatchAndOld.checkpoint(Duration(rddDuration.toInt * 1000))

      val sortRdd = newBatchAndOld.map {
        case (pid, list) => (pid, list.sortBy(_.timestamp))
      }


      sortRdd.foreachRDD(mess =>
        if (!mess.isEmpty()) {
          val nowTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss")
          val todayTimeStamp = DateUtil.getTodayDateTime("yyyyMMdd")
          val yesterdayFormat = DateUtil.getYesterdayFormat()
          val todayFormat = DateUtil.getTimeStampFormat(todayTimeStamp)

          if (nowTimeStamp - todayTimeStamp >= writeFileST.toInt * 60 * 1000 &&
            nowTimeStamp - todayTimeStamp <= writeFileET.toInt * 60 * 1000) {
            println("____________   indicator nowTimeStamp:" + nowTimeStamp)
            val temp = mess.filter(_._2.size >= 2)
            if (!temp.isEmpty()) {
              temp.map { case (pid, list) =>
                val sumSugar = sum2nd(list)
                val SD = sqrt(list.map { a => pow(a.value - sumSugar / list.size, 2) }.sum / (list.size - 1))
                val BGid = list.tail.zip(list).map { a => abs(a._1.value - a._2.value) }
                val exceedSD = BGid.filter(a => a > SD)
                val MAGE = exceedSD.map { a => a / list.size }.sum
                val normal_list = list.filter { a => a.value < 10 && a.value > 3.9 }
                val normal_rate = normal_list.size.toDouble / list.size.toDouble * 100
                val LowSugar_rate_index = list.filter(a => a.value <= 3.9).size.toDouble / list.size.toDouble * 100
                val max_value = list.maxBy(_.value).value
                val mean = sumSugar / list.size

                s"$pid,$yesterdayFormat,$normal_rate,$max_value,$mean,$MAGE,$LowSugar_rate_index"
              }
                .saveAsTextFile(
                  output_indicator_path + todayFormat + "/blood_sugar_indicator/"
                )
            }
          }

          mess.foreachPartition(lines => {
            val nowTimeStamp = DateUtil.getTodayDateTime("yyyyMMddHHmmss")
            val oneHourAgo = nowTimeStamp - 60 * 60 * 1000
            val mqMsg = ArrayBuffer[String]()
            val redisIndicator = ArrayBuffer[(String, String)]()
            val gson = new Gson()
            lines.foreach { case (pid, list) => {
              println("____ pid :" + pid + ",,, list:" + list.map(gson.toJson(_)).toString())

              val lastHourList = list.filter(vo => vo.timestamp > oneHourAgo)
              if (lastHourList.size > 1) {
                val lastBloodSugar: BloodSugarRecord = lastHourList.last
                var preVo = lastHourList.head
                for (i <- 1 to lastHourList.size - 1) {
                  val nowVo = lastHourList(i)
                  if (nowVo.timestamp - preVo.timestamp < 15 * 60 * 1000) {
                    //  @todo 11.低糖:连续两个点血糖<2.8mmol/L:low_glycemia:每5min:触发及时发送. 每小时提示不超过1条
                    if (preVo.value < 2.8 && nowVo.value < 2.8) {
                      val msg = new WarningMsg(
                        "glycemia",
                        lastBloodSugar.personId,
                        NoticeMsg.low_glycemia._1,
                        NoticeMsg.low_glycemia._2,
                        NoticeMsg.low_glycemia._3,
                        preVo.timestamp.toString,
                        NoticeMsg.low_glycemia._5,
                        NoticeMsg.low_glycemia._4,
                        NoticeMsg.low_glycemia._6
                      )
                      mqMsg += gson.toJson(msg)
                    }
                    //  @todo 12.高糖:连续两个点血糖>12.0mmol/L:high_glycemia:每5min:触发及时发送. 每小时提示不超过1条
                    if (preVo.value > 12 && nowVo.value > 12) {
                      val msg = new WarningMsg(
                        "glycemia",
                        lastBloodSugar.personId,
                        NoticeMsg.high_glycemia._1,
                        NoticeMsg.high_glycemia._2,
                        NoticeMsg.high_glycemia._3,
                        preVo.timestamp.toString,
                        NoticeMsg.high_glycemia._5,
                        NoticeMsg.high_glycemia._4,
                        NoticeMsg.high_glycemia._6
                      )
                      mqMsg += gson.toJson(msg)
                    }
                    //@todo 30.空腹高糖:6:00&前后两个点血糖>8.6mmol/L:high_glycemia_fast:每天1次:触发及时发送
                    if (preVo.timestamp >= DateUtil.getTodayATime(5, 45) && nowVo.timestamp <= DateUtil.getTodayATime(6, 15)
                      && preVo.value > 8.6 && nowVo.value > 8.6) {
                      val msg = new WarningMsg(
                        "glycemia",
                        lastBloodSugar.personId,
                        NoticeMsg.high_glycemia_fast._1,
                        NoticeMsg.high_glycemia_fast._2,
                        NoticeMsg.high_glycemia_fast._3,
                        preVo.timestamp.toString,
                        NoticeMsg.high_glycemia_fast._5,
                        NoticeMsg.high_glycemia_fast._4,
                        NoticeMsg.high_glycemia_fast._6
                      )
                      mqMsg += gson.toJson(msg)
                    }
                    preVo = nowVo
                  }
                }
              }

              val sixList = list.filter(vo => vo.timestamp >= DateUtil.getTodayATime(5, 45) && vo.timestamp <= DateUtil.getTodayATime(6, 15))
              if (sixList.size > 1) {
                var preVo = sixList.head
                for (i <- 1 to sixList.size - 1) {
                  val nowVo = sixList(i)
                  if (nowVo.timestamp - preVo.timestamp < 15 * 60 * 1000) {
                    if (preVo.value > 8.6 && nowVo.value > 8.6) {
                      val msg = new WarningMsg(
                        "glycemia",
                        preVo.personId,
                        NoticeMsg.high_glycemia_fast._1,
                        NoticeMsg.high_glycemia_fast._2,
                        NoticeMsg.high_glycemia_fast._3,
                        preVo.timestamp.toString,
                        NoticeMsg.high_glycemia_fast._5,
                        NoticeMsg.high_glycemia_fast._4,
                        NoticeMsg.high_glycemia_fast._6
                      )
                      mqMsg += gson.toJson(msg)
                    }
                  }
                  preVo = nowVo
                }
              }

              //@todo 14.升糖过速，15.降糖过速:血糖上升/下降变化超过6.0mmol/L/h&全天:slope_glycemia_high/slope_glycemia_low:每5min:触发及时发送. 每小时提示不超过1条
              lastHourList.foreach { sugarvo =>
                val thisLastSugar = list.filter(
                  a =>
                    a.timestamp < sugarvo.timestamp && a.timestamp > sugarvo.timestamp - 60 * 60 * 1000
                )
                if (thisLastSugar.nonEmpty) {
                  val lastHourListMax = thisLastSugar.maxBy(_.value)
                  val lastHourListMin = thisLastSugar.minBy(_.value)
                  if (sugarvo.value - lastHourListMin.value > 6) {
                    val msg = new WarningMsg(
                      "glycemia",
                      sugarvo.personId,
                      NoticeMsg.slope_glycemia_high._1,
                      NoticeMsg.slope_glycemia_high._2,
                      NoticeMsg.slope_glycemia_high._3,
                      sugarvo.timestamp.toString,
                      NoticeMsg.slope_glycemia_high._5,
                      NoticeMsg.slope_glycemia_high._4,
                      NoticeMsg.slope_glycemia_high._6
                    )
                    mqMsg += gson.toJson(msg)
                  }
                  if (lastHourListMax.value - sugarvo.value > 6) {
                    val msg = new WarningMsg(
                      "glycemia",
                      sugarvo.personId,
                      NoticeMsg.slope_glycemia_low._1,
                      NoticeMsg.slope_glycemia_low._2,
                      NoticeMsg.slope_glycemia_low._3,
                      sugarvo.timestamp.toString,
                      NoticeMsg.slope_glycemia_low._5,
                      NoticeMsg.slope_glycemia_low._4,
                      NoticeMsg.slope_glycemia_low._6
                    )
                    mqMsg += gson.toJson(msg)
                  }
                }
              }

              //@todo 23.夜间低糖:检测到连续2点<2.8mmol/L&0:00~6:00:low_glycemia_night:每天1次:第二天早上8点回顾性检测0:00~6:00数据
              if (DateUtil.getTodayDateTime() >= DateUtil.getTodayATime(8, 0) && DateUtil
                .getTodayDateTime() < DateUtil.getTodayATime(8, duration.toInt / 60)) {
                val zeroAndSixList = list.filter { vo => vo.timestamp >= DateUtil.getTodayATime(0, 0) && vo.timestamp <= DateUtil.getTodayATime(6, 0) }
                if (zeroAndSixList.size > 1) {
                  var preVo = zeroAndSixList.head
                  for (i <- 1 to zeroAndSixList.size - 1) {
                    val nowVo = zeroAndSixList(i)
                    if (preVo.value < 2.8 && nowVo.value < 2.8
                      && nowVo.timestamp - preVo.timestamp < 15 * 60 * 60 * 1000) {
                      val msg = new WarningMsg(
                        "glycemia",
                        nowVo.personId,
                        NoticeMsg.low_glycemia_night._1,
                        NoticeMsg.low_glycemia_night._2,
                        NoticeMsg.low_glycemia_night._3,
                        nowVo.timestamp.toString,
                        NoticeMsg.low_glycemia_night._5,
                        NoticeMsg.low_glycemia_night._4,
                        NoticeMsg.low_glycemia_night._6
                      )
                      mqMsg += gson.toJson(msg)
                    }
                    preVo = nowVo
                  }
                }
              }

              //@todo  25.午餐前低糖:连续两个点血糖<2.8mmol/L&既定午餐时间前1小时内:low_glycemia_lunch:每次午餐前:触发及时发送
              if (DateUtil.getTodayDateTime() >= DateUtil.getTodayATime(12, 30) && DateUtil
                .getTodayDateTime() < DateUtil.getTodayATime(12, 30 + duration.toInt / 60)) {
                val lunchList = list.filter { vo => vo.timestamp >= DateUtil.getTodayATime(11, 30) && vo.timestamp <= DateUtil.getTodayATime(12, 30) }
                if (lunchList.size > 1) {
                  var preVo = lunchList.head
                  var flag = true
                  for (i <- 1 to lunchList.size - 1) {
                    val nowVo = lunchList(i)
                    if (preVo.value < 2.8 && nowVo.value < 2.8
                      && nowVo.timestamp - preVo.timestamp < 15 * 60 * 60 * 1000 && flag) {
                      flag = false
                      val msg = new WarningMsg(
                        "glycemia",
                        pid.toLong,
                        NoticeMsg.low_glycemia_lunch._1,
                        NoticeMsg.low_glycemia_lunch._2,
                        NoticeMsg.low_glycemia_lunch._3,
                        preVo.timestamp.toString,
                        NoticeMsg.low_glycemia_lunch._5,
                        NoticeMsg.low_glycemia_lunch._4,
                        NoticeMsg.low_glycemia_lunch._6
                      )
                      mqMsg += gson.toJson(msg)
                    }
                    preVo = nowVo
                  }
                }
              }


              //@todo  26.晚餐前低糖:连续两个点血糖<2.8mmol/L&既定晚餐时间前2小时内:low_glycemia_supper:每次晚餐前:触发及时发送
              if (DateUtil.getTodayDateTime() >= DateUtil.getTodayATime(19, 0) && DateUtil
                .getTodayDateTime() < DateUtil.getTodayATime(19, duration.toInt / 60)) {
                val dinnerList = list.filter { vo => vo.timestamp >= DateUtil.getTodayATime(17, 0) && vo.timestamp <= DateUtil.getTodayATime(19, 0) }
                if (dinnerList.size > 1) {
                  var preVo = dinnerList.head
                  var flag = true
                  for (i <- 1 to dinnerList.size - 1) {
                    val nowVo = dinnerList(i)
                    if (preVo.value < 2.8 && nowVo.value < 2.8
                      && nowVo.timestamp - preVo.timestamp < 15 * 60 * 60 * 1000 && flag) {
                      flag = false
                      val msg = new WarningMsg(
                        "glycemia",
                        pid.toLong,
                        NoticeMsg.low_glycemia_supper._1,
                        NoticeMsg.low_glycemia_supper._2,
                        NoticeMsg.low_glycemia_supper._3,
                        preVo.timestamp.toString,
                        NoticeMsg.low_glycemia_supper._5,
                        NoticeMsg.low_glycemia_supper._4,
                        NoticeMsg.low_glycemia_supper._6
                      )
                      mqMsg += gson.toJson(msg)
                    }
                    preVo = nowVo
                  }
                }
              }

              // 血糖数据缺失4个
              // @todo 7.连续4个小时没有血糖数据:连续血糖仪数据中断超过4小时:miss_glycemia_0:每5min:检测到数据中断达到4小时。确认sensor 失效即停止所有CGM相关提示
              if (list.nonEmpty && nowTimeStamp - list.last.timestamp >= 4 * 60 * 60 * 1000
                && nowTimeStamp - list.last.timestamp < 4 * 60 * 60 * 1000 + duration.toInt * 1000) {
                val msg = new WarningMsg(
                  "glycemia",
                  list.last.personId,
                  NoticeMsg.miss_glycemia_0._1,
                  NoticeMsg.miss_glycemia_0._2,
                  NoticeMsg.miss_glycemia_0._3,
                  list.last.timestamp.toString,
                  NoticeMsg.miss_glycemia_0._5,
                  NoticeMsg.miss_glycemia_0._4,
                  NoticeMsg.miss_glycemia_0._6
                )
                mqMsg += gson.toJson(msg)
              }
              // @todo 8.连续7.5个小时没有血糖数据:连续血糖仪数据中断超过7.5小时:miss_glycemia_1:每5min:检测到数据中断达到7.5小时，确认sensor 失效即停止所有CGM相关提示
              if (list.nonEmpty && nowTimeStamp - list.last.timestamp - 7.5 * 60 * 60 * 1000 >= 0
                && nowTimeStamp - list.last.timestamp < 7.5 * 60 * 60 * 1000 + duration.toInt * 1000) {
                val msg = new WarningMsg(
                  "glycemia",
                  list.last.personId,
                  NoticeMsg.miss_glycemia_1._1,
                  NoticeMsg.miss_glycemia_1._2,
                  NoticeMsg.miss_glycemia_1._3,
                  list.last.timestamp.toString,
                  NoticeMsg.miss_glycemia_1._5,
                  NoticeMsg.miss_glycemia_1._4,
                  NoticeMsg.miss_glycemia_1._6
                )
                mqMsg += gson.toJson(msg)
              }
              // @todo 9.连续20个小时没有血糖数据:连续血糖数据中断超20小时:miss_glycemia_2:每5min:检测到数据中断达到16时， 确认sensor 失效即停止所有CGM相关提示
              if (list.nonEmpty && nowTimeStamp - list.last.timestamp - 20 * 60 * 60 * 1000 >= 0
                && nowTimeStamp - list.last.timestamp < 20 * 60 * 60 * 1000 + duration.toInt * 1000
              ) {
                val msg = new WarningMsg(
                  "glycemia",
                  list.last.personId,
                  NoticeMsg.miss_glycemia_2._1,
                  NoticeMsg.miss_glycemia_2._2,
                  NoticeMsg.miss_glycemia_2._3,
                  list.last.timestamp.toString,
                  NoticeMsg.miss_glycemia_2._5,
                  NoticeMsg.miss_glycemia_2._4,
                  NoticeMsg.miss_glycemia_2._6
                )
                mqMsg += gson.toJson(msg)
              }
              val todayList = list.filter(vo => vo.timestamp >= todayTimeStamp)
              if (todayList.size >= 2) {
                //@todo 实时指标4个
                val sumSugar = sum2nd(todayList)
                val SD =
                  sqrt(todayList.map { a => pow(a.value - sumSugar / todayList.size, 2)
                  }.sum / (todayList.size - 1))
                val BGid =
                  todayList.tail.zip(todayList).map { a => abs(a._1.value - a._2.value)
                  }
                val exceedSD = BGid.filter(a => a > SD)
                val MAGE = exceedSD.map { a => a / todayList.size
                }.sum

                val normal_list = todayList.filter { a => a.value < 10 && a.value > 3.9 }
                val normal_rate = normal_list.size.toDouble / todayList.size.toDouble * 100
                val LowSugar_rate_index = todayList.filter(a => a.value <= 3.9).size.toDouble / todayList.size.toDouble * 100
                println("___ normal_list " + normal_list + "\t todayList " + todayList + "\t normal_rate " + normal_rate + "\t LowSugar_rate_index " + LowSugar_rate_index)
                val msg1 = new IndicatorMsg(
                  "glycemia",
                  todayList.last.personId,
                  NoticeMsg.MAGE_index._1,
                  NoticeMsg.MAGE_index._2,
                  NoticeMsg.MAGE_index._3,
                  DateUtil.getTodayDateTime().toString,
                  NoticeMsg.MAGE_index._5,
                  MAGE,
                  NoticeMsg.MAGE_index._6
                )
                mqMsg += gson.toJson(msg1)
                val msg2 = new IndicatorMsg(
                  "glycemia",
                  todayList.last.personId,
                  NoticeMsg.mean_index._1,
                  NoticeMsg.mean_index._2,
                  NoticeMsg.mean_index._3,
                  DateUtil.getTodayDateTime().toString,
                  NoticeMsg.mean_index._5,
                  sumSugar / todayList.size,
                  NoticeMsg.mean_index._6
                )
                mqMsg += gson.toJson(msg2)
                val msg3 = new IndicatorMsg(
                  "glycemia",
                  todayList.last.personId,
                  NoticeMsg.max_index._1,
                  NoticeMsg.max_index._2,
                  NoticeMsg.max_index._3,
                  DateUtil.getTodayDateTime().toString,
                  NoticeMsg.max_index._5,
                  todayList.maxBy(_.value).value,
                  NoticeMsg.max_index._6
                )
                mqMsg += gson.toJson(msg3)
                val msg4 = new IndicatorMsg(
                  "glycemia",
                  todayList.last.personId,
                  NoticeMsg.normal_rate_index._1,
                  NoticeMsg.normal_rate_index._2,
                  NoticeMsg.normal_rate_index._3,
                  todayList.last.timestamp.toString,
                  NoticeMsg.normal_rate_index._5,
                  normal_rate,
                  NoticeMsg.normal_rate_index._6
                )
                val msg5 = new IndicatorMsg(
                  "glycemia",
                  todayList.last.personId,
                  NoticeMsg.LowSugar_rate_index._1,
                  NoticeMsg.LowSugar_rate_index._2,
                  NoticeMsg.LowSugar_rate_index._3,
                  todayList.last.timestamp.toString,
                  NoticeMsg.LowSugar_rate_index._5,
                  LowSugar_rate_index,
                  NoticeMsg.LowSugar_rate_index._6
                )
                redisIndicator +=((
                  "blood_sugar:" + msg1.person_id.toString + ":" + msg1.code,
                  msg1.value.toString
                  ),
                  (
                    "blood_sugar:" + msg2.person_id.toString + ":" + msg2.code,
                    msg2.value.toString
                    ),
                  (
                    "blood_sugar:" + msg3.person_id.toString + ":" + msg3.code,
                    msg3.value.toString
                    ),
                  (
                    "blood_sugar:" + msg4.person_id.toString + ":" + msg4.code,
                    msg4.value.toString
                    ), (
                  "blood_sugar:" + msg5.person_id.toString + ":" + msg5.code,
                  msg5.value.toString
                  ))

                //定时报警3个
                //@todo 34.高血糖：高于目标值时间占比>50, MAGE<0.4&全天:hight_glycemia_ratio:每天1次:随每日总结发送
                //                val HighSugar = todayList.filter(a => a.value > 7.8)
                //                val LowSugar = todayList.filter(a => a.value < 3.9)
                //                if (DateUtil.getTodayDateTime() >= DateUtil.getTodayATime(22, 0) && DateUtil.getTodayDateTime() <= DateUtil.getTodayATime(22, duration.toInt / 60)) {
                //                  if (HighSugar.size / todayList.size > 0.5 && MAGE < 0.4) {
                //                    val msg = new WarningMsg(
                //                      "glycemia",
                //                      todayList.last.personId,
                //                      NoticeMsg.high_glycemia_ratio._1,
                //                      NoticeMsg.high_glycemia_ratio._2,
                //                      NoticeMsg.high_glycemia_ratio._3,
                //                      HighSugar.last.timestamp.toString,
                //                      NoticeMsg.high_glycemia_ratio._5,
                //                      NoticeMsg.high_glycemia_ratio._4,
                //                      NoticeMsg.high_glycemia_ratio._6
                //                    )
                //                    mqMsg += gson.toJson(msg)
                //                  }
                //@todo  33.低血糖：LLD>0.4&全天:low_glycemia_ratio:每天1次:随每日总结发送
                //                  if (LowSugar.size / todayList.size > 0.4) {
                //                    val msg = new WarningMsg(
                //                      "glycemia",
                //                      todayList.last.personId,
                //                      NoticeMsg.low_glycemia_ratio._1,
                //                      NoticeMsg.low_glycemia_ratio._2,
                //                      NoticeMsg.low_glycemia_ratio._3,
                //                      LowSugar.last.timestamp.toString,
                //                      NoticeMsg.low_glycemia_ratio._5,
                //                      NoticeMsg.low_glycemia_ratio._4,
                //                      NoticeMsg.low_glycemia_ratio._6
                //                    )
                //                    mqMsg += gson.toJson(msg)
                //                  }
                //@todo 35.血糖波动：LLD>0.4&全天:mage:每天1次:随每日总结发送
                //                  if (MAGE > 0.6) {
                //                    val msg = new WarningMsg(
                //                      "glycemia",
                //                      todayList.last.personId,
                //                      NoticeMsg.mage._1,
                //                      NoticeMsg.mage._2,
                //                      NoticeMsg.mage._3,
                //                      DateUtil.getTodayDateTime().toString,
                //                      NoticeMsg.mage._5,
                //                      NoticeMsg.mage._4,
                //                      NoticeMsg.mage._6
                //                    )
                //                    mqMsg += gson.toJson(msg)
                //                  }
                //                }
              }

            }
            }

            if (mqMsg.nonEmpty) {
              val mqChannel = mqConn.connect().createChannel()
              mqMsg.foreach { msg =>
                ConnUtil.sendMsg(mqChannel, exchangeName, mqRouteKey, msg)
              }
              mqChannel.close()
            }
            if (redisIndicator.nonEmpty) {
              val jedis = redisConn.connect()
              redisIndicator.foreach {
                case (pid, value) =>
                  jedis.setex(pid, duration.toInt * 2, value)
                  println("____ pid :" + pid + ",value =" + value)
              }
              redisConn.returnResource(jedis)
            }
          })
        })
      ssc
    }

    def sum2nd(aa: ArrayBuffer[BloodSugarRecord]) = {
      aa.map { a => a.value
      }.sum
    }


    val ssc = StreamingContext.getOrCreate(checkpointDir, creatingFunc)

    ssc.start()
    ssc.awaitTermination()
  }
}
